package com.butterfly.hotkey.common.disruptor.container;


import com.butterfly.hotkey.core.constant.Constants;
import com.butterfly.hotkey.common.disruptor.DisruptorProperties;
import com.butterfly.hotkey.common.disruptor.annatation.EventProcessor;
import com.butterfly.hotkey.common.disruptor.processor.IQueueEventProcessor;
import com.butterfly.hotkey.common.disruptor.publish.QueueEventPublisher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.aop.framework.AopProxyUtils;
import org.springframework.aop.scope.ScopedProxyUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.support.GenericApplicationContext;

import java.util.Map;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;

/**
 * @author butterfly
 * @version 1.0
 */
public class QueueProcessorContainerConfiguration implements ApplicationContextAware, SmartInitializingSingleton {

    private static final Logger log = LoggerFactory.getLogger(Constants.WORKER);

    private ConfigurableApplicationContext applicationContext;
    private final AtomicLong counter = new AtomicLong(0);
    private final DisruptorProperties disruptorProperties;
    private final QueueEventPublisher queueEventPublisher;

    public QueueProcessorContainerConfiguration(DisruptorProperties disruptorProperties, QueueEventPublisher queueEventPublisher) {
        this.disruptorProperties = disruptorProperties;
        this.queueEventPublisher = queueEventPublisher;
    }

    @Override
    public void afterSingletonsInstantiated() {
        // 扫描处理器 IQueueEventProcessor, 获取所有的topic,每个队列初始化一份 IQueueEventProcessorContainer
        Map<String, Object> beans = this.applicationContext.getBeansWithAnnotation(EventProcessor.class)
                .entrySet().stream().filter(entry -> !ScopedProxyUtils.isScopedTarget(entry.getKey()))
                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));

        beans.forEach(this::registerContainer);
    }


    private void registerContainer(String beanName, Object bean) {
        Class<?> clazz = AopProxyUtils.ultimateTargetClass(bean);
        if (!IQueueEventProcessor.class.isAssignableFrom(bean.getClass())) {
            throw new IllegalStateException(clazz + " is not instance of " + IQueueEventProcessor.class.getName());
        }

        EventProcessor annotation = clazz.getAnnotation(EventProcessor.class);
        String topic = annotation.topic();
            addContainer(bean, topic);
    }

    private void addContainer(Object bean, String topic) {
        IEventProcessorContainer container = createQueueEventProcessorContainer(bean);
        queueEventPublisher.addContainer(topic, container);
    }

    private IEventProcessorContainer createQueueEventProcessorContainer(Object bean) {
        IEventProcessorContainer container;
        String containerBeanName;
        GenericApplicationContext genericApplicationContext = (GenericApplicationContext) this.applicationContext;
        containerBeanName = String.format("%s_%s", DisruptorProcessorContainer.class.getName(),
                counter.incrementAndGet());
        genericApplicationContext.registerBean(containerBeanName, DisruptorProcessorContainer.class, () -> createDisruptorContainer(containerBeanName, bean));
        container = genericApplicationContext.getBean(containerBeanName, DisruptorProcessorContainer.class);

        if (!container.isRunning()) {
            try {
                container.start();
            } catch (Exception e) {
                log.error("Started container failed. {}", container, e);
                throw new RuntimeException(e);
            }
        }

        return container;
    }

    private DisruptorProcessorContainer createDisruptorContainer(String containerBeanName, Object bean) {
        return new DisruptorProcessorContainer(containerBeanName, disruptorProperties, (IQueueEventProcessor) bean);
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = (ConfigurableApplicationContext) applicationContext;
    }

}
